package com.ibm.mongo;

import java.net.UnknownHostException;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.bson.types.ObjectId;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.IntNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.interfaces.INoSqlDB;
import com.ibm.interfaces.IResultCursor;
import com.ibm.util.DebugUtil;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.LocalConfig;
import com.ibm.util.exception.CPSException;
import com.mongodb.BasicDBList;
import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBObject;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import com.mongodb.MongoException;
import com.mongodb.WriteResult;

/**
 * @author Yaoliang Chen
 * @version May 26, 2014 The mongodb operator class
 */

public class MongoNoSqlDB implements INoSqlDB {
	/**
	 * Reference for log4j
	 */
	private static Logger logger = Logger.getLogger(MongoNoSqlDB.class);

	private MongoClient mongoClient;
	protected DB database;
	protected HashMap<String, Integer> collectionMap;
	protected DBCollection[] collectionSet;
	protected int currentCollection;
	protected String databaseName;
	protected static final long WAIT_TIME = 30000;
	protected final static int MAX_COLLECTION_NUM = 256;

	public MongoNoSqlDB(String uri, String dbName) {
		collectionMap = new HashMap<String, Integer>();
		collectionSet = new DBCollection[MAX_COLLECTION_NUM];
		currentCollection = 0;
		databaseName = dbName;
		while (true) {
			try {
				init(uri, dbName);
				logger.info("===== connected to mongo db =====");
				logger.info(DebugUtil.getStackTrace());
				break;
			} catch (CPSException e) {
				logger.error("fail to init mongo db. wait for " + (WAIT_TIME / 1000.) + " seconds to retry.");
				try {
					Thread.sleep(WAIT_TIME);
				} catch (InterruptedException ie) {
					logger.error(ie.getMessage());
				}
			}
		}
	}

	public MongoNoSqlDB(String hostname, int port, String dbName) {
		collectionMap = new HashMap<String, Integer>();
		collectionSet = new DBCollection[MAX_COLLECTION_NUM];
		currentCollection = 0;
		databaseName = dbName;
		while (true) {
			try {
				init(hostname, port, dbName);
				logger.info("===== connected to mongo db =====");
				logger.info(DebugUtil.getStackTrace());
				break;
			} catch (CPSException e) {
				logger.error("fail to init mongo db. wait for " + (WAIT_TIME / 1000.) + " seconds to retry.");
				try {
					Thread.sleep(WAIT_TIME);
				} catch (InterruptedException ie) {
					logger.error(ie.getMessage());
				}
			}
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.INoSqlDB#init(java.lang.String, int,
	 * java.lang.String)
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.tmp#init(java.lang.String, int,
	 * java.lang.String)
	 */
	@Override
	public void init(String hostname, int port, String dbName) throws CPSException {
		try {
			logger.info("Connecting mongodb uri: " + LocalConfig.MONGO_CONNECT_STRING);
			mongoClient = new MongoClient(hostname, port);
		} catch (UnknownHostException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"Error while connecting nosqldb. \n" + e.getMessage());
		}
		database = mongoClient.getDB(dbName);
		if (database == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "Error when try to open the database " + dbName);
		}
	}

	@Override
	public void init(String uri, String dbName) throws CPSException {
		try {
			logger.info("Connecting mongodb uri: " + uri);
			mongoClient = new MongoClient(new MongoClientURI(uri));
		} catch (UnknownHostException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"Error while connecting nosqldb. \n" + e.getMessage());
		}
		if (dbName == null) {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "Please provide mongo database in uri " + uri);
		}
		database = mongoClient.getDB(dbName);
		if (database == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "Error when try to open the database " + dbName);
		}

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.tmp#getCollection(java.lang.String)
	 */
	@Override
	public int getCollection(String name) throws CPSException {
		Integer colid = collectionMap.get(name);
		if (colid != null) {
			return colid;
		}

		if (currentCollection == MAX_COLLECTION_NUM - 1) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "Too many collection opened.");
		}

		DBCollection col = database.getCollection(name);
		if (col == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "No such collection " + name);
		}
		collectionMap.put(name, currentCollection);
		collectionSet[currentCollection++] = col;
		return currentCollection - 1;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.tmp#insert(int,
	 * com.fasterxml.jackson.databind.node.ObjectNode, java.lang.String,
	 * java.util.HashSet)
	 */
	@Override
	public void insert(int colid, ObjectNode json, String keyTag, HashSet<String> tags) throws CPSException {
		if (colid < 0 || colid >= MAX_COLLECTION_NUM || collectionSet[colid] == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"collection does not exist in nosql database. id  " + colid + " does not exist. ");
		}

		DBObject bson = encode(json, keyTag, tags);
		try {
			collectionSet[colid].insert(bson);
		} catch (MongoException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}
	}

	@Override
	public void createIndex(int colid, String indexTag) throws CPSException {
		if (colid < 0 || colid >= MAX_COLLECTION_NUM || collectionSet[colid] == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"collection does not exist in nosql database. id  " + colid + " does not exist. ");
		}

		try {
			collectionSet[colid].createIndex(indexTag);
		} catch (MongoException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}
	}

	@Override
	public void createIndex(int colid, String[] indexTag) throws CPSException {
		if (colid < 0 || colid >= MAX_COLLECTION_NUM || collectionSet[colid] == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"collection does not exist in nosql database. id  " + colid + " does not exist. ");
		}

		DBObject node = new BasicDBObject();
		for (int i = 0; i < indexTag.length; i++) {
			node.put(indexTag[i], 1);
		}

		try {
			collectionSet[colid].createIndex(node, new ObjectId().toHexString());
		} catch (MongoException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}
	}

	@Override
	public void createUniqueIndex(int colid, String indexTag) throws CPSException {
		if (colid < 0 || colid >= MAX_COLLECTION_NUM || collectionSet[colid] == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"collection does not exist in nosql database. id  " + colid + " does not exist. ");
		}

		DBObject key = new BasicDBObject(indexTag, 1);

		try {
			collectionSet[colid].createIndex(key, null, true);
		} catch (MongoException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}
	}

	@Override
	public void upsert(int colid, ObjectNode json, String keyTag, HashSet<String> tags) throws CPSException {
		if (colid < 0 || colid >= MAX_COLLECTION_NUM || collectionSet[colid] == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"collection does not exist in nosql database. id  " + colid + " does not exist. ");
		}

		DBObject bson = encode(json, keyTag, tags);
		DBObject key = new BasicDBObject("_id", bson.get("_id"));

		try {
			collectionSet[colid].update(key, bson, true, false);
		} catch (MongoException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.INoSqlDB#findOne(int, java.lang.Object)
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.tmp#findOne(int, long, java.lang.String[])
	 */
	@Override
	public ObjectNode findOne(int collid, Object _id, String keyTag) throws CPSException {
		if (collid < 0 || collid >= MAX_COLLECTION_NUM || collectionSet[collid] == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"collection does not exist in nosql database. id  " + collid + " does not exist. ");
		}
		ObjectNode node = null;
		DBObject obj = null;
		try {
			obj = collectionSet[collid].findOne(_id);
			if (obj == null) {
				/* no doc with _id is found */
				return null;
			}
			node = (ObjectNode) JSONUtility.newJSONNode(obj.toMap());
			node.set(keyTag, node.remove("_id"));
			return node;
		} catch (Exception e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.INoSqlDB#find(int, org.json.ObjectNode)
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.tmp#find(int,
	 * com.fasterxml.jackson.databind.node.ObjectNode, boolean,
	 * java.lang.String)
	 */
	@Override
	public IResultCursor<ObjectNode> find(int collid, ObjectNode filter, boolean onlyKey, String tagName)
			throws CPSException {
		if (collid < 0 || collid >= MAX_COLLECTION_NUM || collectionSet[collid] == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"collection does not exist in nosql database. id  " + collid + " does not exist. ");
		}

		DBObject ref = null;
		if (filter != null) {
			ref = encode(filter, tagName, true);
			if (ref == null)
				return null;
		}
		try {
			if (onlyKey) {
				return new MongoResultCursor(collectionSet[collid].find(ref, IDProjection), tagName);
			} else {
				return new MongoResultCursor(collectionSet[collid].find(ref), tagName);
			}
		} catch (MongoException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}

	}

	static final private BasicDBObject IDProjection = new BasicDBObject("_id", 1);

	private DBObject encode(ArrayNode a, String keyTag) throws CPSException {
		if (a == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "null json array");
		}
		BasicDBList result = new BasicDBList();
		try {
			for (int i = 0; i < a.size(); ++i) {
				JsonNode o = a.get(i);
				if (o instanceof ObjectNode) {
					result.add(encode((ObjectNode) o, keyTag, false));
				} else if (o instanceof ArrayNode) {
					result.add(encode((ArrayNode) o, keyTag));
				} else {
					result.add(o.asText());
				}
			}
			return result;
		} catch (Exception e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}
	}

	private DBObject encode(ObjectNode o, String keyTag, HashSet<String> tags) throws CPSException {
		if (tags == null)
			return encode(o, keyTag, true);
		if (o == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "null json doc");
		}
		BasicDBObject result = new BasicDBObject();
		try {
			Iterator<Entry<String, JsonNode>> i = o.fields();
			while (i.hasNext()) {
				Entry<String, JsonNode> node = i.next();
				String k = (String) node.getKey();
				JsonNode v = node.getValue();
				if (!tags.contains(k)) {
					throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "unexpected tag '" + k + "'");
				}
				if (k.equals(keyTag))
					k = "_id";
				if (v.isArray()) {
					result.put(k, encode((ArrayNode) v, keyTag));
				} else if (v.isObject()) {
					result.put(k, encode((ObjectNode) v, keyTag, false));
				} else {
					result.put(k, JSONUtility.getObjectFromNode(v));
				}
			}
			return result;
		} catch (Exception e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}
	}

	private DBObject encode(ObjectNode o, String keyTag, boolean key2id) throws CPSException {
		if (o == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "null json doc");
		}

		BasicDBObject result = new BasicDBObject();
		try {
			Iterator<Entry<String, JsonNode>> i = o.fields();
			while (i.hasNext()) {
				Entry<String, JsonNode> node = i.next();
				String k = (String) node.getKey();
				JsonNode v = node.getValue();
				if (key2id && k.equals(keyTag))
					k = "_id";
				if (v.isArray()) {
					result.put(k, encode((ArrayNode) v, keyTag));
				} else if (v.isObject()) {
					result.put(k, encode((ObjectNode) v, keyTag, false));
				} else {
					result.put(k, JSONUtility.getObjectFromNode(v));
				}
			}
			return result;
		} catch (Exception e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.tmp#update(int,
	 * com.fasterxml.jackson.databind.node.ObjectNode, java.lang.String,
	 * java.util.HashSet)
	 */
	@Override
	public int update(int collid, ObjectNode doc, String keyTag, HashSet<String> tags) throws CPSException {
		if (collid < 0 || collid >= MAX_COLLECTION_NUM || collectionSet[collid] == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"collection does not exist in nosql database. id  " + collid + " does not exist. ");
		}
		WriteResult wr = null;
		try {
			long _id = doc.remove(keyTag).asLong();
			if (_id == 0) {
				throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "invalid doc key");
			}

			DBObject _key = new BasicDBObject("_id", _id);
			DBObject _doc = encode(doc, keyTag, tags);
			wr = collectionSet[collid].update(_key, _doc);
			doc.put(keyTag, _id);

			return wr.getN();
		} catch (CPSException e) {
			throw e;
		} catch (Exception e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.tmp#delete(int, long, java.lang.String[])
	 */
	@Override
	public int delete(int collid, Object key) throws CPSException {
		if (collid < 0 || collid >= MAX_COLLECTION_NUM || collectionSet[collid] == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"collection does not exist in nosql database. id  " + collid + " does not exist. ");
		}
		DBObject searchKey = new BasicDBObject("_id", key);
		try {
			WriteResult wr = collectionSet[collid].remove(searchKey);
			return wr.getN();
		} catch (Exception e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.tmp#getDB()
	 */
	@Override
	public Object getDB() {
		return mongoClient;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.tmp#getCollection(int, java.lang.String[])
	 */
	@Override
	public Object getCollection(int collection) throws CPSException {
		if (collection >= collectionSet.length) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"no such collection in nosql database. id=" + collection);
		}
		return collectionSet[collection];
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.tmp#dropDatabase()
	 */
	@Override
	public void dropDatabase() throws CPSException {
		try {
			mongoClient.dropDatabase(databaseName);
		} catch (MongoException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.WhaleDB.tools.mongo.tmp#bulkInsert(int, java.util.List,
	 * java.lang.String, java.util.HashSet)
	 */
	@Override
	public void bulkInsert(int colid, List<ObjectNode> jsons, String keyTag, HashSet<String> tags) throws CPSException {
		if (colid < 0 || colid >= MAX_COLLECTION_NUM || collectionSet[colid] == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"collection does not exist in nosql database. id  " + colid + " does not exist. ");
		}
		DBObject[] bsons = new DBObject[jsons.size()];
		for (int i = 0; i < jsons.size(); i++) {
			DBObject bson = encode(jsons.get(i), keyTag, tags);
			if (bson == null) {
				throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "null json doc");
			}
			bsons[i] = bson;
		}

		try {
			WriteResult wr = collectionSet[colid].insert(bsons);
			if (bsons.length != wr.getN()) {
				throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "not all the schemas are inserted.");
			}
		} catch (MongoException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, e.getMessage());
		}

	}

	@Override
	public void flush() throws CPSException {
		// do nothing
	}

	public void testBasicObject() {
		BasicDBObject object = new BasicDBObject("key", "value").append("key2", "value2");
	}

	@Override
	public void deleteCollections(Collection<String> collNames) {
		for (String collection : collNames) {
			database.getCollection(collection).drop();
		}
	}

	public static void main(String[] args) {
		BasicDBList result = new BasicDBList();
		JsonNode o = new IntNode(1);
		result.add(o.asText());
		result.add(o.toString());
		System.out.println(result);
	}

	@Override
	public void close() throws CPSException {
		mongoClient.close();
		logger.info("===== disconnected from mongo db =====");
		logger.info(DebugUtil.getStackTrace());
	}

}
